弹力设计篇之"幂等性设计" - [2026重制版]

public class SnowflakeIdGenerator {
    // ==================== 常量 ====================
    /** 开始时间截 (2021-01-01) */
    private final long twepoch = 1609430400000L;

    /** 机器id所占的位数 */
    private final long workerIdBits = 5L;

    /** 数据标识id所占的位数 */
    private final long datacenterIdBits = 5L;

    /** 支持的最大机器id，结果是31 (这个移位算法可以很快计算出几位二进制数所能表示的最大十进制数) */
    private final long maxWorkerId = -1L ^ (-1L << workerIdBits);

    /** 支持的最大数据标识id，结果是31 */
    private final long maxDatacenterId = -1L ^ (-1L << datacenterIdBits);

    /** 序列在id中占的位数 */
    private final long sequenceBits = 12L;

    /** 机器ID向左移12位 */
    private final long workerIdShift = sequenceBits;

    /** 数据标识id向左移17位(12+5) */
    private final long datacenterIdShift = sequenceBits + workerIdBits;

    /** 时间戳向左移22位(5+5+12) */
    private final long timestampLeftShift = sequenceBits + workerIdBits + datacenterIdBits;

    /** 生成序列的掩码，这里为4095 (0b111111111111=0xfff=4095) */
    private final long sequenceMask = -1L ^ (-1L << sequenceBits);

    /** 工作机器ID(0~31) */
    private long workerId;

    /** 数据中心ID(0~31) */
    private long datacenterId;

    /** 毫秒内序列(0~4095) */
    private long sequence = 0L;

    /** 上次生成ID的时间截 */
    private long lastTimestamp = -1L;

    //==============================构造函数=========================
    /**
     * 构造函数
     * @param workerId 工作ID (0~31)
     * @param datacenterId 数据中心ID (0~31)
     */
    public SnowflakeIdGenerator(long workerId, long datacenterId) {
        if (workerId > maxWorkerId || workerId < 0) {
            throw new IllegalArgumentException(String.format("worker Id can't be greater than %d or less than 0", maxWorkerId));
        }
        if (datacenterId > maxDatacenterId || datacenterId < 0) {
            throw new IllegalArgumentException(String.format("datacenter Id can't be greater than %d or less than 0", maxDatacenterId));
        }
        this.workerId = workerId;
        this.datacenterId = datacenterId;
    }

    // ==============================方法=================================
    /**
     * 获得下一个ID (该方法是线程安全的)
     * @return SnowflakeId
     */
    public synchronized long nextId() {
        long timestamp = timeGen();

        //如果当前时间小于上一次ID生成的时间戳，说明系统时钟回退过这个时候应当抛出异常
        if (timestamp < lastTimestamp) {
            throw new RuntimeException(
                String.format("Clock moved backwards. Refusing to generate id for %d milliseconds", lastTimestamp - timestamp));
        }

        //如果是同一时间生成的，则进行毫秒内序列
        if (lastTimestamp == timestamp) {
            sequence = (sequence + 1) & sequenceMask;
            //毫秒内序列溢出
            if (sequence == 0) {
                //阻塞到下一个毫秒,获得新的时间戳
                timestamp = tilNextMillis(lastTimestamp);
            }
        }
        //时间戳改变，毫秒内序列重置
        else {
            sequence = 0L;
        }

        //上次生成ID的时间截
        lastTimestamp = timestamp;

        //移位并通过或运算拼到一起组成64位的ID
        return ((timestamp - twepoch) << timestampLeftShift) //
            | (datacenterId << datacenterIdShift) //
            | (workerId << workerIdShift) //
            | sequence;
    }

    /**
     * 阻塞到下一个毫秒，直到获得新的时间戳
     * @param lastTimestamp 上次生成ID的时间截
     * @return 当前时间戳
     */
    protected long tilNextMillis(long lastTimestamp) {
        long timestamp = timeGen();
        while (timestamp <= lastTimestamp) {
            timestamp = timeGen();
        }
        return timestamp;
    }

    /** 返回以毫秒为单位的当前时间 */
    protected long timeGen() {
        return System.currentTimeMillis();
    }
}

// UUID v7 Java实现 (使用java.util.UUID或第三方库)
import com.fasterxml.uuid.Generators;
import com.fasterxml.uuid.impl.TimeBasedEpochGenerator;

// 使用UUID v7生成有序且唯一的ID
UUID uuid7 = Generators.timeBasedEpochGenerator().generate();

// 输出示例: 018f3b3d-5b8e-7c9d-e0f1-23456789abcd
// 特点:
// - 前48位是Unix毫秒时间戳（有序）
// - 后74位是随机数（保证唯一性）
// - 可以直接存入数据库的VARCHAR(36)或转换为BIGINT

@Service
@RequiredArgsConstructor
@Slf4j
public class RedisIdempotencyService {

    private final StringRedisTemplate redisTemplate;

    /** 幂等key前缀 */
    private static final String IDEMPOTENCY_KEY_PREFIX = "idempotency:";

    /** 默认过期时间：24小时 */
    private static final Duration DEFAULT_TTL = Duration.ofHours(24);

    /**
     * 检查并记录幂等性
     *
     * @param idempotencyKey 幂等键（如请求ID）
     * @param ttl 过期时间
     * @return true 表示首次请求，false 表示重复请求
     */
    public boolean checkAndSet(String idempotencyKey, Duration ttl) {
        String key = IDEMPOTENCY_KEY_PREFIX + idempotencyKey;
        
        // SETNX: 只有当key不存在时才设置，原子操作
        Boolean isFirstRequest = redisTemplate.opsForValue()
            .setIfAbsent(key, "1", ttl != null ? ttl : DEFAULT_TTL);
        
        return Boolean.TRUE.equals(isFirstRequest);
    }

    /**
     * 检查并记录幂等性（带结果缓存）
     *
     * @param idempotencyKey 幂等键
     * @param result 业务执行结果（JSON字符串）
     * @param ttl 过期时间
     * @return IdempotencyResult 包装对象
     */
    public <T> IdempotencyResult<T> checkAndSetWithResult(
            String idempotencyKey,
            T result,
            Duration ttl) {

        String key = IDEMPOTENCY_KEY_PREFIX + idempotencyKey;
        
        // Lua脚本保证原子性：检查是否存在，存在则返回旧值，不存在则设置新值
        String luaScript =
            "if redis.call('exists', KEYS[1]) == 1 then " +
            "   return redis.call('get', KEYS[1]) " +
            "else " +
            "   redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) " +
            "   return nil " +
            "end";

        DefaultRedisScript<String> script = new DefaultRedisScript<>();
        script.setScriptText(luaScript);
        script.setResultType(String.class);

        String cachedResult = redisTemplate.execute(
            script,
            Collections.singletonList(key),
            toJson(result),
            String.valueOf(ttl.getSeconds())
        );

        if (cachedResult != null) {
            // 之前已经处理过，返回缓存的结果
            T previousResult = fromJson(cachedResult, (Class<T>) result.getClass());
            return IdempotencyResult.duplicate(previousResult);
        } else {
            // 首次处理
            return IdempotencyResult.firstTime(result);
        }
    }

    /**
     * 删除幂等记录（慎用！仅在补偿/回滚场景下使用）
     */
    public void delete(String idempotencyKey) {
        String key = IDEMPOTENCY_KEY_PREFIX + idempotencyKey;
        redisTemplate.delete(key);
    }
}

/**
 * 幂等性检查结果包装类
 */
@Data
@Builder
public class IdempotencyResult<T> {
    private boolean firstRequest;      // 是否首次请求
    private T result;                  // 执行结果（首次或缓存）
    private boolean duplicate;         // 是否重复请求

    public static <T> IdempotencyResult<T> firstTime(T result) {
        return IdempotencyResult.<T>builder()
            .firstRequest(true)
            .result(result)
            .duplicate(false)
            .build();
    }

    public static <T> IdempotencyResult<T> duplicate(T cachedResult) {
        return IdempotencyResult.<T>builder()
            .firstRequest(false)
            .result(cachedResult)
            .duplicate(true)
            .build();
    }
}

@RestController
@RequestMapping("/api/orders")
@RequiredArgsConstructor
public class OrderController {

    private final OrderService orderService;
    private final RedisIdempotencyService idempotencyService;

    /**
     * 创建订单（带幂等性保护）
     */
    @PostMapping
    public ResponseEntity<OrderResponse> createOrder(
            @RequestBody CreateOrderRequest request,
            @RequestHeader(value = "X-Idempotency-Key", required = false) String idempotencyKey) {

        // 1. 生成或获取幂等key
        if (StringUtils.isBlank(idempotencyKey)) {
            idempotencyKey = UUID.randomUUID().toString();
        }

        // 2. 幂等性检查
        IdempotencyResult<OrderResponse> idempotencyResult = 
            idempotencyService.checkAndSet(idempotencyKey, Duration.ofHours(24));

        if (idempotencyResult.isDuplicate()) {
            // 重复请求，直接返回之前的结果
            log.info("Duplicate order request detected, returning cached result");
            return ResponseEntity.ok()
                .header("X-Idempotency-Key", idempotencyKey)
                .header("X-Cache-Hit", "true")
                .body(idempotencyResult.getResult());
        }

        // 3. 首次请求，执行业务逻辑
        try {
            OrderResponse response = orderService.createOrder(request);
            
            // 4. 更新幂等缓存中的结果
            idempotencyService.checkAndSetWithResult(idempotencyKey, response, Duration.ofHours(24));

            return ResponseEntity.created(URI.create("/api/orders/" + response.getOrderId()))
                .header("X-Idempotency-Key", idempotencyKey)
                .header("X-Cache-Hit", "false")
                .body(response);

        } catch (Exception e) {
            // 发生异常，删除幂等key允许重试（可选策略）
            // idempotencyService.delete(idempotencyKey);
            throw e;
        }
    }
}

-- 创建幂等性记录表
CREATE TABLE idempotent_records (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    idempotency_key VARCHAR(64) NOT NULL COMMENT '幂等键（唯一）',
    business_type VARCHAR(32) NOT NULL COMMENT '业务类型',
    business_id VARCHAR(64) NOT NULL COMMENT '业务主键ID',
    request_payload JSON COMMENT '请求参数快照',
    response_payload JSON COMMENT '响应结果缓存',
    status TINYINT NOT NULL DEFAULT 0 COMMENT '状态: 0-处理中 1-成功 2-失败',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    expire_at TIMESTAMP COMMENT '过期时间',
    
    UNIQUE KEY uk_idempotency_key (idempotency_key),
    INDEX idx_business (business_type, business_id),
    INDEX idx_expire (expire_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 
COMMENT='幂等性记录表';

@Mapper
public interface IdempotentRecordMapper {

    /**
     * 尝试插入幂等记录（利用唯一约束保证原子性）
     * @return 影响行数（1表示成功，0表示已存在）
     */
    @Insert("INSERT INTO idempotent_records " +
            "(idempotency_key, business_type, business_id, request_payload, status, expire_at) " +
            "VALUES (#{key}, #{businessType}, #{businessId}, #{requestPayload}, #{status}, #{expireAt})")
    int insert(@Param("key") String idempotencyKey,
               @Param("businessType") String businessType,
               @Param("businessId") String businessId,
               @Param("requestPayload") String requestPayload,
               @Param("status") int status,
               @Param("expireAt") LocalDateTime expireAt);

    /**
     * 更新响应结果
     */
    @Update("UPDATE idempotent_records " +
            "SET response_payload = #{responsePayload}, status = #{status}, updated_at = NOW() " +
            "WHERE idempotency_key = #{key}")
    int updateResponse(@Param("key") String idempotencyKey,
                       @Param("responsePayload") String responsePayload,
                       @Param("status") int status);

    /**
     * 查询幂等记录
     */
    @Select("SELECT * FROM idempotent_records WHERE idempotency_key = #{key}")
    Optional<IdempotentRecord> findByKey(@Param("key") String idempotencyKey);
}

@Service
@Transactional
@RequiredArgsConstructor
@Slf4j
public class DatabaseIdempotencyService {

    private final IdempotentRecordMapper recordMapper;

    /**
     * 执行幂等操作
     */
    public <T> IdempotencyResult<T> executeIdempotent(
            String idempotencyKey,
            String businessType,
            Supplier<T> operation,
            Class<T> resultType) {

        // 1. 尝试插入幂等记录
        int inserted = recordMapper.insert(
            idempotencyKey,
            businessType,
            null,  // businessId稍后更新
            null,  // requestPayload
            0,     // PROCESSING状态
            LocalDateTime.now().plusHours(24)  // 24小时后过期
        );

        if (inserted == 0) {
            // 2. 已存在，查询之前的记录
            Optional<IdempotentRecord> existing = recordMapper.findByKey(idempotencyKey);
            if (existing.isPresent()) {
                IdempotentRecord record = existing.get();
                if (record.getStatus() == 1) {  // SUCCESS
                    T cachedResult = fromJson(record.getResponsePayload(), resultType);
                    log.info("Returning cached result for key: {}", idempotencyKey);
                    return IdempotencyResult.duplicate(cachedResult);
                } else if (record.getStatus() == 0) {  // PROCESSING
                    // 正在处理中，可能发生了并发请求或长时间运行的任务
                    log.warn("Request is still processing for key: {}", idempotencyKey);
                    throw new RequestProcessingException("Request is being processed");
                } else {  // FAILED
                    log.error("Previous execution failed for key: {}", idempotencyKey);
                    throw new PreviousExecutionFailedException(record);
                }
            }
        }

        // 3. 首次执行，调用实际业务逻辑
        try {
            T result = operation.get();

            // 4. 更新幂等记录为成功状态
            recordMapper.updateResponse(idempotencyKey, toJson(result), 1);  // SUCCESS

            return IdempotencyResult.firstTime(result);

        } catch (Exception e) {
            // 5. 更新幂等记录为失败状态
            recordMapper.updateResponse(idempotencyKey, toJson(e.getMessage()), 2);  // FAILED
            
            log.error("Execution failed for idempotent key: {}", idempotencyKey, e);
            throw e;
        }
    }
}

-- 订单表（包含业务唯一约束）
CREATE TABLE orders (
    order_id VARCHAR(32) PRIMARY KEY COMMENT '订单ID（幂等键）',
    user_id VARCHAR(32) NOT NULL,
    total_amount DECIMAL(10, 2) NOT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
    version INT NOT NULL DEFAULT 1 COMMENT '乐观锁版本号',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

    -- 业务约束：防止同一用户在同一时间段内重复下单（可选）
    UNIQUE KEY uk_user_order_time (user_id, created_at(100))
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- MySQL: INSERT ... ON DUPLICATE KEY UPDATE
INSERT INTO orders (order_id, user_id, total_amount, status)
VALUES ('ORD-20260606-001', 'USER-789', 199.98, 'CREATED')
ON DUPLICATE KEY UPDATE
    updated_at = NOW();  -- 不做任何实质性修改，只是触发唯一约束检查

-- PostgreSQL: INSERT ... ON CONFLICT DO NOTHING
INSERT INTO orders (order_id, user_id, total_amount, status)
VALUES ('ORD-20260606-001', 'USER-789', 199.98, 'CREATED')
ON CONFLICT (order_id) DO NOTHING;

@Service
@RequiredArgsConstructor
public class OrderService {

    private final OrderMapper orderMapper;

    /**
     * 创建订单（利用唯一索引实现幂等性）
     */
    @Transactional(rollbackFor = Exception.class)
    public Order createOrder(CreateOrderRequest request) {
        // 1. 生成订单ID作为幂等键
        String orderId = generateOrderId();  // Snowflake/UUID v7
        
        // 2. 构建订单对象
        Order order = Order.builder()
            .orderId(orderId)
            .userId(request.getUserId())
            .totalAmount(request.getTotalAmount())
            .status(OrderStatus.CREATED)
            .build();

        try {
            // 3. 尝试插入（如果orderId已存在会抛出DuplicateKeyException）
            orderMapper.insert(order);
            log.info("Order created successfully: {}", orderId);
            return order;

        } catch (DuplicateKeyException e) {
            // 4. 主键冲突，说明订单已存在，查询并返回
            log.warn("Order already exists: {}", orderId);
            return orderMapper.selectById(orderId);
        }
    }
}

# OpenAPI/Swagger 定义（展示幂等性相关字段）
paths:
  /api/orders:
    post:
      summary: 创建订单（非幂等，需配合Idempotency-Key使用）
      description: |
        创建新订单。
        
        ## 幂等性保障
        此接口通过 `Idempotency-Key` 请求头实现幂等性。
        客户端应在第一次请求时生成唯一的Idempotency-Key，
        并在后续的重试请求中使用相同的值。
        
        服务端会缓存首次请求的响应，并在检测到相同的Idempotency-Key时
        直接返回缓存的响应，不会重复执行业务逻辑。
      
      parameters:
        - name: Idempotency-Key
          in: header
          required: true
          schema:
            type: string
            format: uuid
            example: "550e8400-e29b-41d4-a716-446655440000"
          description: |
            幂等性键（必须全局唯一）
            推荐使用 UUID v4/v7 或 ULID
      
      responses:
        '201':
          description: 订单创建成功
          headers:
            Idempotency-Key:
              schema:
                type: string
              description: 返回原始的幂等性键
            X-Cache-Hit:
              schema:
                type: boolean
              description: 是否命中缓存（true表示这是重复请求）
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/OrderResponse'
        
        '200':
          description: 重复请求，返回缓存的响应
          headers:
            X-Cache-Hit:
              schema:
                type: boolean
              enum: [true]
              description: 始终为true（表示命中了幂等性缓存）
          
        '409':
          description: 冲突（Idempotency-Key已被用于不同的请求内容）
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Error'

// TypeScript/JavaScript 客户端示例
import { v4 as uuidv4 } from 'uuid';
import axios from 'axios';

class IdempotentApiClient {
  private idempotencyKeys: Map<string, string> = new Map();

  /**
   * 发送幂等的POST请求
   */
  async post<T>(url: string, data: any, config?: AxiosRequestConfig): Promise<T> {
    const cacheKey = `${url}:${JSON.stringify(data)}`;
    
    // 获取或生成幂等性Key
    let idempotencyKey = this.idempotencyKeys.get(cacheKey);
    if (!idempotencyKey) {
      idempotencyKey = uuidv4();  // 或使用 crypto.randomUUID()
      this.idempotencyKeys.set(cacheKey, idempotencyKey);
      
      // 设置自动清理（例如30分钟后）
      setTimeout(() => this.idempotencyKeys.delete(cacheKey), 30 * 60 * 1000);
    }

    try {
      const response = await axios.post(url, data, {
        ...config,
        headers: {
          ...config?.headers,
          'Idempotency-Key': idempotencyKey,
        },
      });

      // 检查是否命中了服务端缓存
      const cacheHit = response.headers['x-cache-hit'] === 'true';
      if (cacheHit) {
        console.log(`[Idempotency] Cached response returned for ${url}`);
      }

      return response.data;

    } catch (error: any) {
      // 对于可重试的错误（如网络超时），保持Key以便下次重试
      if (this.isRetryableError(error)) {
        console.log(`[Idempotency] Keeping key for retry: ${idempotencyKey}`);
        throw error;  // 让上层重试逻辑处理
      } else {
        // 不可重试的错误（如400 Bad Request），清除Key
        this.idempotencyKeys.delete(cacheKey);
        throw error;
      }
    }
  }

  private isRetryableError(error: any): boolean {
    if (!error.response) {
      return true;  // 网络错误，可重试
    }
    const status = error.response.status;
    // 5xx服务器错误、408请求超时、429太多请求
    return status >= 500 || status === 408 || status === 429;
  }
}

@Service
@Transactional
@RequiredArgsConstructor
@Slf4j
public class PaymentService {

    private final PaymentRepository paymentRepo;
    private final RedisIdempotencyService idempotencyService;
    private final PaymentGatewayClient gatewayClient;

    /**
     * 发起支付（对外接口）
     */
    public PaymentResult initiatePayment(PaymentRequest request, String idempotencyKey) {
        // 1. Redis快速幂等检查（毫秒级）
        IdempotencyResult<PaymentResult> redisCheck = 
            idempotencyService.checkAndSetWithResult(
                idempotencyKey, 
                null, 
                Duration.ofMinutes(30)
            );
        
        if (redisCheck.isDuplicate()) {
            log.info("Payment already in progress, returning cached status");
            return redisCheck.getResult() != null ? 
                redisCheck.getResult() : 
                PaymentResult.processing(request.getPaymentId());
        }

        // 2. 数据库层面幂等保障（利用唯一约束）
        try {
            Payment payment = Payment.builder()
                .paymentId(request.getPaymentId())       // 业务ID作为唯一键
                .orderId(request.getOrderId())
                .amount(request.getAmount())
                .currency(request.getCurrency())
                .status(PaymentStatus.INITIATED)
                .idempotencyKey(idempotencyKey)           // 保存幂等key便于追踪
                .build();
            
            paymentRepo.save(payment);  // 可能抛出DuplicateKeyException
            
            // 3. 调用外部支付渠道
            GatewayResponse gatewayResp = gatewayClient.charge(request);
            
            // 4. 更新支付状态
            payment.setStatus(PaymentStatus.PROCESSING);
            payment.setGatewayTransactionId(gatewayResp.getTransactionId());
            paymentRepo.save(payment);
            
            // 5. 缓存中间状态到Redis（供快速查询）
            PaymentResult result = PaymentResult.processing(payment.getPaymentId());
            idempotencyService.checkAndSetWithResult(idempotencyKey, result, Duration.ofMinutes(30));
            
            return result;

        } catch (DuplicateKeyException e) {
            // 支付记录已存在，查询当前状态返回
            Payment existing = paymentRepo.findByPaymentId(request.getPaymentId())
                .orElseThrow(() -> new PaymentNotFoundException(request.getPaymentId()));
            
            log.warn("Payment record already exists: {} with status: {}", 
                     request.getPaymentId(), existing.getStatus());
                     
            return mapToResult(existing);
            
        } catch (Exception e) {
            log.error("Failed to initiate payment: {}", request.getPaymentId(), e);
            // 清除Redis缓存允许客户端重试
            idempotencyService.delete(idempotencyKey);
            throw new PaymentFailedException(e);
        }
    }

    /**
     * 处理支付渠道回调（确保幂等性）
     */
    public void handleGatewayCallback(GatewayCallback callback) {
        String transactionId = callback.getTransactionId();
        
        // 查找对应的支付记录
        Payment payment = paymentRepo.findByGatewayTransactionId(transactionId)
            .orElseThrow(() -> new PaymentNotFoundException(transactionId));

        // 幂等性检查：只有特定状态的支付才能更新
        if (!canUpdateStatus(payment.getStatus())) {
            log.info("Ignoring callback for payment {} with current status: {}", 
                     payment.getPaymentId(), payment.getStatus());
            return;  // 忽略重复或无效的回调
        }

        // 更新支付状态
        if (callback.isSuccess()) {
            payment.setStatus(PaymentStatus.SUCCESS);
            payment.setPaidAt(LocalDateTime.now());
        } else {
            payment.setStatus(PaymentStatus.FAILED);
            payment.setFailureReason(callback.getErrorMessage());
        }
        
        paymentRepo.save(payment);
        
        // 发布支付完成事件（异步通知下游系统）
        eventPublisher.publishEvent(new PaymentCompletedEvent(payment));
    }

    private boolean canUpdateStatus(PaymentStatus currentStatus) {
        return currentStatus == PaymentStatus.PROCESSING 
            || currentStatus == PaymentStatus.INITIATED;
    }
}

-- 定时任务清理过期记录（每天执行一次）
DELETE FROM idempotent_records 
WHERE expire_at < NOW() - INTERVAL 7 DAY 
AND status IN (1, 2);  -- 只清理已完成或失败的记录